package bigdata.hermesfuxi.eagle.etl.jobs;

import bigdata.hermesfuxi.eagle.etl.bean.DataLogBean;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.util.Collector;

import java.util.*;

/**
 * LiveGiftPointStatistics 简化版本 - 直接使用Map进行礼物信息关联
 */
@Slf4j
public class LiveGiftPointStatisticsSimple {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setMaxParallelism(128);
        env.setParallelism(1);

        // 创建礼物信息映射
        Map<Integer, Tuple2<String, Integer>> giftMap = new HashMap<>();
        giftMap.put(1, Tuple2.of("玫瑰", 10));
        giftMap.put(2, Tuple2.of("跑车", 100));
        giftMap.put(3, Tuple2.of("飞机", 500));
        giftMap.put(4, Tuple2.of("火箭", 1000));
        giftMap.put(5, Tuple2.of("城堡", 5000));

        // 创建测试数据流
        DataStream<DataLogBean> beanStream = env.fromElements(
            createTestData("liveReward", "anchor001", 1),
            createTestData("liveReward", "anchor002", 2),
            createTestData("liveReward", "anchor001", 3),
            createTestData("liveReward", "anchor003", 1),
            createTestData("liveReward", "anchor002", 4),
            createTestData("liveReward", "anchor001", 5)
        );

        beanStream.print("输入数据");

        // 处理数据，关联礼物信息
        SingleOutputStreamOperator<DataLogBean> processedStream = beanStream.process(new ProcessFunction<DataLogBean, DataLogBean>() {
            @Override
            public void processElement(DataLogBean bean, Context ctx, Collector<DataLogBean> out) throws Exception {
                Integer giftId = bean.getGiftId();
                String anchorId = bean.getAnchorId();
                
                if (giftId != null && anchorId != null && giftMap.containsKey(giftId)) {
                    Tuple2<String, Integer> giftInfo = giftMap.get(giftId);
                    bean.setGiftName(giftInfo.f0);
                    bean.setPoints(giftInfo.f1);
                    
                    log.info("处理数据: anchorId={}, giftId={}, giftName={}, points={}", 
                            anchorId, giftId, giftInfo.f0, giftInfo.f1);
                    
                    out.collect(bean);
                }
            }
        });

        processedStream.print("处理结果");

        // 统计主播积分
        SingleOutputStreamOperator<Tuple2<String, Integer>> anchorPointsStream = processedStream
            .map(new MapFunction<DataLogBean, Tuple2<String, Integer>>() {
                @Override
                public Tuple2<String, Integer> map(DataLogBean bean) throws Exception {
                    return Tuple2.of(bean.getAnchorId(), bean.getPoints());
                }
            })
            .keyBy(t -> t.f0)
            .sum(1);

        anchorPointsStream.print("主播积分统计");

        // 统计礼物数量
        SingleOutputStreamOperator<Tuple2<String, Integer>> giftCountStream = processedStream
            .map(new MapFunction<DataLogBean, Tuple2<String, Integer>>() {
                @Override
                public Tuple2<String, Integer> map(DataLogBean bean) throws Exception {
                    return Tuple2.of(bean.getGiftName(), 1);
                }
            })
            .keyBy(t -> t.f0)
            .sum(1);

        giftCountStream.print("礼物统计");

        // 计算TopN
        SingleOutputStreamOperator<List<Map.Entry<String, Integer>>> topNStream = anchorPointsStream
            .keyBy(t -> "key")
            .process(new TopNProcessFunction(3));

        topNStream.print("主播积分Top3");

        env.execute("LiveGiftPointStatisticsSimple");
    }

    private static DataLogBean createTestData(String eventId, String anchorId, Integer giftId) {
        DataLogBean dataLogBean = new DataLogBean();
        dataLogBean.setEventId(eventId);
        dataLogBean.setDeviceId("device" + System.currentTimeMillis());
        dataLogBean.setTimestamp(System.currentTimeMillis());
        dataLogBean.setAnchorId(anchorId);
        dataLogBean.setGiftId(giftId);
        
        Map<String, Object> properties = new HashMap<>();
        properties.put("anchor_id", anchorId);
        properties.put("gift_id", giftId);
        dataLogBean.setProperties(properties);
        
        return dataLogBean;
    }

    public static class TopNProcessFunction extends KeyedProcessFunction<String, Tuple2<String, Integer>, List<Map.Entry<String, Integer>>> {
        private final int topN;
        private transient MapState<String, Integer> mapState;

        public TopNProcessFunction(int topN) {
            this.topN = topN;
        }

        @Override
        public void open(Configuration parameters) throws Exception {
            mapState = getRuntimeContext().getMapState(new MapStateDescriptor<>("topNState", String.class, Integer.class));
        }

        @Override
        public void processElement(Tuple2<String, Integer> value, Context ctx, Collector<List<Map.Entry<String, Integer>>> out) throws Exception {
            mapState.put(value.f0, value.f1);
            
            // 注册定时器，每10秒输出一次TopN
            long currentTime = ctx.timerService().currentProcessingTime();
            long triggerTime = currentTime - currentTime % 10000 + 10000;
            ctx.timerService().registerProcessingTimeTimer(triggerTime);
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<List<Map.Entry<String, Integer>>> out) throws Exception {
            List<Map.Entry<String, Integer>> allEntries = new ArrayList<>();
            for (Map.Entry<String, Integer> entry : mapState.entries()) {
                allEntries.add(entry);
            }
            
            allEntries.sort((a, b) -> b.getValue().compareTo(a.getValue()));
            
            List<Map.Entry<String, Integer>> topNList = allEntries.subList(0, Math.min(topN, allEntries.size()));
            out.collect(topNList);
        }
    }
}
